package com.atguigu.bigdata.spark

import java.sql.DriverManager

import org.apache.spark.rdd.JdbcRDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.util.parsing.json.JSON

//创建自定义分区
object Spark02_MySql14{

  def main(args: Array[String]): Unit = {
    //创建SparkConf
    //s设定spark计算框架的运行环境
    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("wordCount")
    //创建Spark上下文环境
    val sc = new SparkContext(config)

    //3.定义连接mysql的参数
    val driver = "com.mysql.jdbc.Driver"
    val url = "jdbc:mysql://hadoop102:3306/rdd"
    val userName = "root"
    val passWd = "000000"
//查詢數據
/*    val sql = "select name,age from user where id >=? and id <=?"

    val jdbcRDD = new JdbcRDD(
      sc,
      () => {
        //       获取数据库的连接对象
        Class.forName(driver)
        DriverManager.getConnection(url, userName, passWd)
      },
      sql,
      //      id >=1 and id <=3 分为两个分区
      1,
      3,
      2,
      (rs) => {
        println(rs.getString(1) + "," + rs.getString(2))
      }
    )
    jdbcRDD.collect()*/
    val dataRDD = sc.makeRDD(List(("liliu",20),("wangjiu",25)))
    //   这样子会导致频繁连接
    //
    //    dataRDD.foreach {
    //       case (username, age) => {
    //       Class.forName(driver)
    //       val connection = DriverManager.getConnection(url, userName, passWd)
    //        val sql = "insert into user (name,age) values(?,?)"
    //        val statement = connection.prepareStatement(sql)
    //        statement.setString(1,username)
    //        statement.setInt(2,age)
    //        statement.executeUpdate()
    //
    //        statement.close()
    //
    //      }
    //    }
//   这样子会导致无法序列化
//    Class.forName(driver)
//    val connection = DriverManager.getConnection(url, userName, passWd)
//    dataRDD.foreach {
//      case (username, age) => {
//
//        val sql = "insert into user (name,age) values(?,?)"
//        val statement = connection.prepareStatement(sql)
//        statement.setString(1,username)
//        statement.setInt(2,age)
//        statement.executeUpdate()
//
//        statement.close()
//
//      }
//    }

//效率会提高但是可能会出现内存溢出的现象
    dataRDD.foreachPartition(datas=> {
      Class.forName(driver)
      val connection = DriverManager.getConnection(url, userName, passWd)

      datas.foreach{
             case (username, age) => {

               val sql = "insert into user (name,age) values(?,?)"
               val statement = connection.prepareStatement(sql)
               statement.setString(1,username)
               statement.setInt(2,age)
               statement.executeUpdate()

               statement.close()

             }
     }
      connection.close()
    })

    sc.stop()
  }
}
